package com.eastwit.api.controller.liangshan;

import cn.hutool.core.date.DateTime;
import cn.hutool.core.date.DateUtil;
import cn.hutool.http.Header;
import cn.hutool.http.HttpRequest;
import cn.hutool.http.HttpUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;


/**
 * @description:凉山kafka车辆、人脸数据拉取
 * @author: cyy
 **/
@Component
@Slf4j
public class KafkaConsumerController {

    @Value("${liangshang.NotificationID}")
    private String NotificationID;

    @Value("${liangshang.SubscribeID}")
    private String SubscribeID;

    @Value("${liangshang.faceUrl}")
    private String FaceUrl;

    @Value("${viomUrl}")
    private String viomUrl;

    private final static String MOTO = "moto";

    private final static String Face = "Face";


    @KafkaListener(topics = "${consumer.topic-motor}", groupId = "dz-group", autoStartup = "${kafka.autoStartup}" , concurrency = "2")
    public void motorInfo(ConsumerRecords<Object, String> records, Acknowledgment ack) {
        try {
            JSONObject rootObject = new JSONObject();
            JSONObject secondObject = new JSONObject();
            JSONArray secondArr = new JSONArray();
            JSONObject threeObject = new JSONObject();
            threeObject.put("NotificationID", NotificationID);
            threeObject.put("SubscribeID", SubscribeID);
            threeObject.put("Title", MOTO);
            String nowDate = DateUtil.parse(DateUtil.now()).toString("yyyyMMddHHmmss");
            threeObject.put("TriggerTime", nowDate);
            threeObject.put("InfoIDs", "");
            JSONArray motorVehicleObjectArr = new JSONArray();
            log.info("read kafka motorInfo start");
            for (ConsumerRecord record : records) {
                JSONObject motorVehicleObject = new JSONObject();
                String value = String.valueOf(record.value());
                JSONObject motorJson = JSONObject.parseObject(value);
                String StorageUrl1 = String.valueOf(motorJson.get("StorageUrl1"));
                String DeviceID = String.valueOf(motorJson.get("DeviceID"));
                motorVehicleObject.put("MotorVehicleID", String.valueOf(motorJson.get("MotorVehicleID")));
                motorVehicleObject.put("StorageUrl1", StorageUrl1);
                motorVehicleObject.put("LaneNo", Integer.parseInt(String.valueOf(motorJson.get("LaneNo"))));
                motorVehicleObject.put("VehicleClass", String.valueOf(motorJson.get("VehicleClass")));
                motorVehicleObject.put("PassTime", String.valueOf(motorJson.get("PassTime")));
                motorVehicleObject.put("SafetyBelt", 1);
                motorVehicleObject.put("PlateReliability", String.valueOf(motorJson.get("PlateReliability")));
                motorVehicleObject.put("BrandReliability", String.valueOf(motorJson.get("BrandReliability")));
                motorVehicleObject.put("InfoKind", Integer.parseInt(String.valueOf(motorJson.get("InfoKind"))));
                motorVehicleObject.put("SourceID", String.valueOf(motorJson.get("SourceID")));
                motorVehicleObject.put("DeviceID", String.valueOf(motorJson.get("DeviceID")));
                motorVehicleObject.put("HasPlate", String.valueOf(motorJson.get("HasPlate")));
                motorVehicleObject.put("PlateClass", String.valueOf(motorJson.get("PlateClass")));
                motorVehicleObject.put("PlateColor", String.valueOf(motorJson.get("PlateColor")));
                motorVehicleObject.put("PlateNo", String.valueOf(motorJson.get("PlateNo")));
                motorVehicleObject.put("VehicleBrand", String.valueOf(motorJson.get("VehicleBrand")));
                motorVehicleObject.put("VehicleColor", String.valueOf(motorJson.get("VehicleColor")));
                JSONArray subImageInfoObjectArr = new JSONArray();
                JSONObject subImageInfoObject = new JSONObject();
                subImageInfoObject.put("ImageID", "");
                subImageInfoObject.put("EventSort", 12);
                subImageInfoObject.put("DeviceID", DeviceID);
                subImageInfoObject.put("StoragePath", StorageUrl1);
                subImageInfoObject.put("FileFormat", "Jpeg");
                subImageInfoObject.put("ShotTime", nowDate);
                subImageInfoObject.put("Width", 0);
                subImageInfoObject.put("Height", 0);
                subImageInfoObject.put("Type", "01");
                subImageInfoObjectArr.add(subImageInfoObject);
                JSONObject jsonSub = new JSONObject();
                jsonSub.put("SubImageInfoObject", subImageInfoObjectArr);
                motorVehicleObject.put("SubImageList", jsonSub);
                motorVehicleObjectArr.add(motorVehicleObject);
                System.out.println("result:" + value);
            }
            JSONObject MotorVehicleObject = new JSONObject();
            MotorVehicleObject.put("MotorVehicleObject", motorVehicleObjectArr);
            threeObject.put("MotorVehicleObjectList", MotorVehicleObject);
            secondArr.add(threeObject);
            secondObject.put("SubscribeNotificationObject", secondArr);
            rootObject.put("SubscribeNotificationListObject", secondObject);
            log.info("read kafka motorInfo finish");
            String result = HttpUtil.createPost(viomUrl)
                    .charset("UTF-8")
                    .header(Header.CONTENT_TYPE, "application/json")
                    .body(rootObject.toString())
                    .execute().body();
            JSONObject responseStatusObject = (JSONObject) JSONObject.parseObject(result).get("ResponseStatusListObject");
            JSONArray responseStatusArr = (JSONArray) responseStatusObject.get("ResponseStatusObject");
            JSONObject jsonObject = responseStatusArr.getJSONObject(0);
            String statusCode = jsonObject.get("StatusCode").toString();
            String statusString = jsonObject.get("StatusString").toString();
            if ("0".equals(statusCode)) {
                log.info("数据发送成功=====" + statusCode + "=======成功描述:" + statusString + "=====" + nowDate);
            } else {
                log.error("数据发送失败=====" + statusCode + "=======错误描述:" + statusString + "=====" + nowDate);
            }
        } catch (Exception e) {
            log.error("kafka motor data error", e);
            log.error("data:{}", records);
            ack.acknowledge();
        }
        ack.acknowledge();
    }


    @KafkaListener(topics = "#{'${consumer.topic-Faces}'.split(',')}", groupId = "dz-group1", autoStartup = "${kafka.autoStartup}",concurrency = "2")
    public void zhxqFacesInfo(ConsumerRecords<Object, String> records, Acknowledgment ack) {
        try {
            JSONObject rootObject = new JSONObject();
            JSONObject secondObject = new JSONObject();
            JSONArray secondArr = new JSONArray();
            JSONObject threeObject = new JSONObject();
            threeObject.put("NotificationID", NotificationID);
            threeObject.put("SubscribeID", SubscribeID);
            threeObject.put("Title", Face);
            String nowDate = DateUtil.parse(DateUtil.now()).toString("yyyyMMddHHmmss");
            threeObject.put("TriggerTime", nowDate);
            threeObject.put("InfoIDs", "");
            JSONArray faceObjectArr = new JSONArray();
            log.info("read kafka zhxqFacesInfo start");
            for (ConsumerRecord record : records) {
                //too many parameters are useless and unnecessary
                //i'm a rookie don't spray me
                JSONObject faceObject = new JSONObject();
                String value = String.valueOf(record.value());
                JSONObject jsonObject = JSONObject.parseObject(value);
                String DeviceID = String.valueOf(jsonObject.get("DeviceID"));
                String bigImagePath = String.valueOf(jsonObject.get("BigImagePath"));
                String BigImagePath = FaceUrl + bigImagePath;
                faceObject.put("FaceID", "");
                faceObject.put("SourceID", "");
                faceObject.put("ShotTime", nowDate);
                faceObject.put("DeviceID", DeviceID);
                faceObject.put("InfoKind", 1);
                faceObject.put("LeftTopX", 0);
                faceObject.put("LeftTopX", 0);
                faceObject.put("RightBtmX", nowDate);
                faceObject.put("RightBtmY", nowDate);
                faceObject.put("LocationMarkTime", 50);
                faceObject.put("AgeUpLimit", 50);
                faceObject.put("IsDriver", 2);
                faceObject.put("IsForeigner", 2);
                faceObject.put("IsSuspectedTerrorist", 2);
                faceObject.put("IsCriminalInvolved", 2);
                faceObject.put("IsDetainees", 2);
                faceObject.put("IsVictim", 2);
                faceObject.put("IsSuspiciousPerson", 2);
                JSONArray subImageInfoObjectArr = new JSONArray();
                JSONObject subImageInfoObject = new JSONObject();
                subImageInfoObject.put("ImageID", "");
                subImageInfoObject.put("DeviceID", DeviceID);
                subImageInfoObject.put("StoragePath", BigImagePath);
                subImageInfoObject.put("Type", "11");
                subImageInfoObject.put("FileFormat", "Jpeg");
                subImageInfoObject.put("ShotTime", nowDate);
                subImageInfoObject.put("Width", 1);
                subImageInfoObject.put("Height", 1);
                subImageInfoObjectArr.add(subImageInfoObject);
                JSONObject jsonSub = new JSONObject();
                jsonSub.put("SubImageInfoObject", subImageInfoObjectArr);
                faceObject.put("SubImageList", jsonSub);
                faceObjectArr.add(faceObject);
            }
            JSONObject FaceObject = new JSONObject();
            FaceObject.put("FaceObject", faceObjectArr);
            threeObject.put("FaceObjectList", FaceObject);
            secondArr.add(threeObject);
            secondObject.put("SubscribeNotificationObject", secondArr);
            rootObject.put("SubscribeNotificationListObject", secondObject);
            log.info("read kafka zhxqFacesInfo finish");

            String result = HttpUtil.createPost(viomUrl)
                    .charset("UTF-8")
                    .header(Header.CONTENT_TYPE, "application/json")
                    .body(rootObject.toString())
                    .execute().body();
            JSONObject responseStatusObject = (JSONObject) JSONObject.parseObject(result).get("ResponseStatusListObject");
            JSONArray responseStatusArr = (JSONArray) responseStatusObject.get("ResponseStatusObject");
            JSONObject jsonObject = responseStatusArr.getJSONObject(0);
            String statusCode = jsonObject.get("StatusCode").toString();
            String statusString = jsonObject.get("StatusString").toString();
            if ("0".equals(statusCode)) {
                log.info("数据发送成功=====" + statusCode + "=======成功描述:" + statusString + "=====" + nowDate);
            } else {
                log.info("数据发送错误=====" + statusCode + "=======错误描述:" + statusString + "=====" + nowDate);
            }
        } catch (Exception e) {
            log.error("kafka zhxq data error", e);
            log.error("data:{}", records);
            ack.acknowledge();
        }
        ack.acknowledge();
    }

}
